/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.utils.serde;

import java.io.IOException;
import java.io.InputStream;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;

import javax.jdo.identity.ObjectIdentity;
import javax.jdo.identity.SingleFieldIdentity;

import org.apache.commons.lang.StringUtils;
import org.datanucleus.ObjectManager;
import org.datanucleus.api.ApiAdapter;
import org.datanucleus.exceptions.NucleusDataStoreException;
import org.datanucleus.metadata.AbstractClassMetaData;
import org.datanucleus.metadata.AbstractMemberMetaData;
import org.datanucleus.store.ObjectProvider;
import cn.edu.thu.laud.objectstore.index.CompositeUtil;
import cn.edu.thu.laud.utils.serde.Serializer;
import org.datanucleus.store.mapped.exceptions.DatastoreFieldDefinitionException;
import org.datanucleus.store.types.ObjectLongConverter;
import org.datanucleus.store.types.ObjectStringConverter;
import org.datanucleus.store.types.TypeManager;

/**
 * Context per Store Manager that allows conversion of data types to native
 * bytes. Stored the mappings to allow users to define their own conversion
 * scheme. The converter uses the default byte encoding for all Java primitives.
 * All Strings are encode with UTF 8. To create your own converters, define a
 * properties file like the following and put it into your classpath
 * 
 * FileName: cassandraByteCoverters.properties
 * 
 * Format:
 * 
 * <pre>
 * <classname>:<converter classname>
 * 
 * </pre>
 * 
 * Then in your JDO configuration add the following properties
 * 
 * <pre>
 * <?xml version="1.0" encoding="utf-8"?>
 * <jdoconfig xmlns="http://java.sun.com/xml/ns/jdo/jdoconfig"
 * 	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 * 	xsi:noNamespaceSchemaLocation="http://java.sun.com/xml/ns/jdo/jdoconfig">
 * 
 * 	<!-- Datastore Txn PMF -->
 * 	<persistence-manager-factory name="Test">
 * 		<property name="javax.jdo.PersistenceManagerFactoryClass"
 * 			value="org.datanucleus.jdo.JDOPersistenceManagerFactory" />
 * 		<property name="javax.jdo.option.ConnectionURL" value="cassandra://keyspace/host:port,..." />
 * 		<property name="javax.jdo.option.Optimistic" value="false" />
 * 		
 * 		<!--  auto creates keyspace and CF.  Not recommended for production environments since replication factor and strategy are defaulted to 1 and simple-->
 * 		<property name="datanucleus.autoCreateSchema" value="true" />
 * 		
 * 		<property name="datanucleus.autoCreateTables" value="true"/>
 * 		
 * 		<property name="datanucleus.autoCreateColumns" value="true"/>
 * 			
 * 		<property name="cn.edu.thu.laud.utils.serde.serializer" value="cn.edu.thu.laud.utils.serde.JavaSerializer"/>
 * 		
 * 		<property name="cn.edu.thu.laud.utils.serde.bytemapper" value="cassandraByteCoverters.properties"/>
 * 
 * 	</persistence-manager-factory>
 * 
 * </jdoconfig>
 * </pre>
 * 
 * Note that each converter must implement the @see
 * 
 * @author zhuoan
 * 
 */
public class ConverterContext {

	private Map<Class<?>, IConverter> converters;

	private IConverter boolConverter;
	private IConverter charConverter;
	private IConverter shortConverter;
	private IConverter intConverter;
	private IConverter doubleConverter;
	private IConverter longConverter;
	private IConverter floatConverter;
	private IConverter stringConverter;
	private IConverter serializerConverter;

	private Serializer serializer;
	private TypeManager typeManager;
	private ApiAdapter apiAdapter;

	public ConverterContext(String propertiesFilePath,
			Serializer serializer, TypeManager typeManager,
			ApiAdapter apiAdapter) {
		this.serializer = serializer;
		this.typeManager = typeManager;
		this.apiAdapter = apiAdapter;
		initialize(propertiesFilePath);

	}

	/**
	 * Initialize
	 * 
	 * @param propertiesFilePath
	 */
	private void initialize(String propertiesFilePath) {
		
		converters = new HashMap<Class<?>, IConverter>();

		/**
		 * Load our defaults
		 */
		converters.put(Boolean.class, new BooleanConverter());
		converters.put(Character.class, new CharacterConverter());
		converters.put(Double.class, new DoubleConverter());
		converters.put(Float.class, new FloatConverter());
		converters.put(Integer.class, new IntegerConverter());
		converters.put(java.util.UUID.class, new LexicalUUIDConverter());
		converters.put(Long.class, new LongConverter());
		converters.put(Short.class, new ShortConverter());
		converters.put(String.class, new StringConverter());
		converters.put(com.eaio.uuid.UUID.class, new TimeUUIDConverter());
		converters.put(byte[].class, new ByteArrayConverter());

		// Load user defined props and override defaults if required
		if (propertiesFilePath != null) {

			InputStream propsFileSource = this.getClass().getClassLoader()
					.getResourceAsStream(propertiesFilePath);

			if (propsFileSource == null) {
				throw new NucleusDataStoreException(String.format(
						"Could not load properties file %s from classpath",
						propertiesFilePath));
			}

			Properties converterProps = new Properties();
			try {
				converterProps.load(propsFileSource);
			} catch (IOException e) {
				throw new NucleusDataStoreException(String.format(
						"Could not load properties file %s from classpath",
						propertiesFilePath), e);
			}

			for (Entry<Object, Object> converterDef : converterProps.entrySet()) {
				String classKey = converterDef.getKey().toString();
				String converterClassName = converterDef.getValue().toString();

				try {
					IConverter instance = (IConverter) Class.forName(
							converterClassName).newInstance();

					converters.put(Class.forName(classKey), instance);

				} catch (Exception e) {
					throw new NucleusDataStoreException(
							String.format(
									"Unable to instanciate converter for key class %s with converter class %s.  Please make a no arg constructor is present in the converter",
									classKey, converterClassName), e);
				}

			}
		}

		// wire our our pointers to our primitive times after potential override
		this.boolConverter = converters.get(Boolean.class);
		this.charConverter = converters.get(Character.class);
		this.doubleConverter = converters.get(Double.class);
		this.floatConverter = converters.get(Float.class);
		this.intConverter = converters.get(Integer.class);
		this.longConverter = converters.get(Long.class);
		this.shortConverter = converters.get(Short.class);
		this.stringConverter = converters.get(String.class);

		this.serializerConverter = new SerializerWrapperConverter(serializer);

	}

	/**
	 * Convenience method to find an object given a string form of its identity,
	 * and the metadata for the class (or a superclass).
	 * 
	 * @param idStr
	 *            The id string
	 * @param cmd
	 *            Metadata for the class
	 * @return The object
	 */
	public ByteBuffer getRowKey(ObjectProvider op) {

		return getRowKey(op.getObject());

	}
	
	public ByteBuffer getPKValue(ObjectManager om, ObjectProvider op) {

		AbstractClassMetaData acmd = op.getClassMetaData();
		int[] pks = acmd.getPKMemberPositions();

		if (pks == null || pks.length != 1) {
			throw new NucleusDataStoreException(
					"Currently only single field identity objects are allowed");
		}
		
		Object pkVal = op.provideField(pks[0]);
		IConverter converter = determineConverter(pkVal.getClass());
		ByteBuffer pkBytes = convertToBytes(converter, pkVal, null);
		pkBytes.limit(pkBytes.position());
		pkBytes.reset();
		// System.out.println(pkVal+"=" +new String(Hex.encodeHex(ByteBufferUtil.getArray(pkBytes))));
		return pkBytes;

	}

	/**
	 * Get the stringified key for the given execution context and object
	 * 
	 * @param ec
	 * @param object
	 * @return
	 */
	public ByteBuffer getRowKey(Object object) {
		Object id = apiAdapter.getIdForObject(object);

		return getRowKeyForId(id);
	}

	/**
	 * Get the row key for the given id
	 * 
	 * @param ec
	 * @param id
	 * @return
	 */
	public ByteBuffer getRowKeyForId(Object id) {
		ByteBuffer buffer = getRowKeyForId(id, null);
		buffer.limit(buffer.position());
		buffer.reset();
		return buffer;
	} 

	/**
	 * Get the row key for the given id. If the object is just a plain object
	 * I.E not an object identity or single field identity, then it will convert
	 * the bytes using the appropriate converter.
	 * 
	 * @param ec
	 * @param id
	 * @return
	 */
	public ByteBuffer getRowKeyForId(Object id, ByteBuffer buffer) {

		Object objectId = null;

		if (id instanceof ObjectIdentity) {
			objectId = ((ObjectIdentity) id).getKey();

		} else if (id instanceof SingleFieldIdentity) {
			objectId = ((SingleFieldIdentity) id).getKeyAsObject();
		} else {
			objectId = id;
		}

		IConverter converter = determineConverter(objectId.getClass());

		if (converter == serializerConverter) {
			throw new NucleusDataStoreException(
					String.format(
							"You cannot use the default serializer on a key.  See the %s to defined your own converter or %s interface to use custom keys",
							ConverterContext.class.getName(),
							ByteAware.class.getName()));
		}

		return convertToBytes(converter, objectId, buffer);

	}
	
	/**
	 * Convert the value to bytes using the defined converters. Converts in the
	 * following order
	 * <ol>
	 * <li>ByteConverter from converter Mapping</li>
	 * <li>ByteAware implementation</li>
	 * <li>ObjectLongConverter which is cached</li>
	 * <li>ObjectStringConverter which is cached</li>
	 * <li>Serialization using the serializer which is cached</li>
	 * </ol>
	 * 
	 * @param value
	 * @param ec
	 * @return
	 */
	public ByteBuffer getBytes(Object value) {
		
		IConverter converter = converters.get(value.getClass());

		if (converter != null) {
			return convertToBytes(converter, value);
		}

		converter = determineConverter(value.getClass());

		converters.put(value.getClass(), converter);

		return convertToBytes(converter, value);
	}
	
   public Object getIndexValue(Object value) {
		
	   
	   IConverter converter = converters.get(value.getClass());

		if (converter == null) {
			converter = determineConverter(value.getClass());
			converters.put(value.getClass(), converter);
		}
		
		if(converter == serializerConverter)
			return null;

		String validation = converter.getComparatorType();
		if(!(StringUtils.indexOfAny(validation, CompositeUtil.INDEX_TYPES) > -1))
			return null;
        if(value instanceof Number || value instanceof String)
        	return value;
        if(value instanceof com.eaio.uuid.UUID){
        	com.eaio.uuid.UUID uuid = (com.eaio.uuid.UUID)value;
        	return new java.util.UUID(uuid.time, uuid.clockSeqAndNode);
        }
        
        if(validation.equals("IntegerType")){
        	ObjectLongConverter dnLongConverter = typeManager
    				.getLongConverter(value.getClass());
        	return dnLongConverter.toLong(value);
        }
        
        if(validation.equals("UTF8Type")){
        	ObjectStringConverter dnStringConverter = typeManager
    				.getStringConverter(value.getClass());
        	return dnStringConverter.toString(value);
        }
        
        return getBytes(value);
		

		
	}

	/**
	 * Convert to bytes using the given buffer. Does not reset the buffer and
	 * will return the buffer to be used for future operations
	 * 
	 * @param value
	 * @param buffer
	 */
	public ByteBuffer getBytes(Object value, ByteBuffer buffer) {

		IConverter converter = converters.get(value.getClass());

		if (converter == null) {

			converter = determineConverter(value.getClass());

			converters.put(value.getClass(), converter);
		}

		return converter.writeBytes(value, buffer, this);

	}

	/**
	 * Convert the bytes to a value using the defined converters.
	 * 
	 * <ol>
	 * <li>ByteConverter from converter Mapping</li>
	 * <li>ByteAware implementation</li>
	 * <li>ObjectLongConverter which is cached</li>
	 * <li>ObjectStringConverter which is cached</li>
	 * <li>Serialization using the serializer which is cached</li>
	 * </ol>
	 * 
	 * @param value
	 * @return
	 */
	public Object getObject(ByteBuffer value, AbstractClassMetaData meta,
			int fieldNumber) {

		AbstractMemberMetaData member = meta
				.getMetaDataForManagedMemberAtAbsolutePosition(fieldNumber);

		return getObject(value, member.getType());

	}

	/**
	 * Convert the bytes to a value using the defined converters.
	 * 
	 * <ol>
	 * <li>ByteConverter from converter Mapping</li>
	 * <li>ByteAware implementation</li>
	 * <li>ObjectLongConverter which is cached</li>
	 * <li>ObjectStringConverter which is cached</li>
	 * <li>Serialization using the serializer which is cached</li>
	 * </ol>
	 * 
	 * @param value
	 *            The bytes value
	 * @param targetType
	 *            The class type of the target object
	 * @return
	 */
	public Object getObject(ByteBuffer buffer, Class<?> targetType) {
		IConverter converter = converters.get(targetType);

		if (converter != null) {
			return convertToObject(converter, buffer);
		}

		converter = determineConverter(targetType);

		converters.put(targetType, converter);

		return convertToObject(converter, buffer);
	}

	/**
	 * Determine the converter that should be used for this class. Will not
	 * perform any caching on converters that are created, however it does check
	 * the cache for existing definitions Creates converters with the following
	 * order
	 * 
	 * <ol>
	 * <li>ByteConverter from converter Mapping</li>
	 * <li>ByteAware implementation</li>
	 * <li>ObjectLongConverter which is cached</li>
	 * <li>ObjectStringConverter which is cached</li>
	 * <li>Serialization using the serializer which is cached</li>
	 * </ol>
	 * 
	 * @param clazz
	 * @return
	 */
	private IConverter determineConverter(Class<?> clazz) {

		IConverter converter = converters.get(clazz);

		if (converter != null) {
			return converter;
		}

		if (ByteAware.class.isAssignableFrom(clazz)) {
			return new ByteAwareConverter(clazz);
		}

		ObjectLongConverter dnLongConverter = typeManager
				.getLongConverter(clazz);

		if (dnLongConverter != null) {
			return new ObjectLongWrapperConverter(dnLongConverter,
					this.longConverter);
		}

		ObjectStringConverter dnStringConverter = typeManager
				.getStringConverter(clazz);

		if (dnStringConverter != null) {
			return new ObjectStringWrapperConverter(dnStringConverter,
					this.stringConverter);
		}

		return serializerConverter;

	}

	/**
	 * Get the key class (when used with converters) for the given classes meta
	 * data
	 * 
	 * @param ec
	 * @param cmd
	 * @return
	 */
	public Class<?> getKeyClass(ObjectManager om, AbstractClassMetaData cmd) {
		
		ApiAdapter adapter = om.getApiAdapter();

		if (!adapter.isSingleFieldIdentityClass(cmd.getObjectidClass())) {
			throw new DatastoreFieldDefinitionException(
					"multiple field identity creation is currently unsupported in indexing");
		}

		AbstractMemberMetaData member = cmd
				.getMetaDataForManagedMemberAtAbsolutePosition(cmd
						.getPKMemberPositions()[0]);

		return member.getType();
	}

	/**
	 * Object for the identity in the AbstractClassMetaData's class. Try and
	 * build it from the string
	 * 
	 * @param op
	 * @param cmd
	 * @param value
	 * @return
	 */
	public Object getObjectIdentity(ObjectManager om,
			Class<?> candidateClass, ByteBuffer value) {

		AbstractClassMetaData cmd = om.getMetaDataManager()
				.getMetaDataForClass(candidateClass,
						om.getClassLoaderResolver());

		Class<?> keyType = getKeyClass(om, cmd);
		
		try {
			// if the class of the pk is a primitive we'll want to get the value
			// then set it as a string

			Object targetId = getObject(value, keyType);

			return om.newObjectId(candidateClass, targetId);

		} catch (Exception e) {
			throw new NucleusDataStoreException(
					"Unable to serialize bytes to object identity.  Please make sure it has the same SerializationId, long or string converter is was stored with. ");
		}

	}

	/**
	 * Return the cassandra validator for the class specified. Using the
	 * converter to determine the type. See determineConverter to view the rules
	 * for creating a converter.
	 * 
	 * @param fieldClass
	 * @return
	 * @see determineConverter
	 */
	public String getValidationClass(Class<?> fieldClass) {

		IConverter converter = determineConverter(fieldClass);

		return converter.getComparatorType();
	}

	public ByteBuffer getBytes(Boolean value) {
		return convertToBytes(this.boolConverter, value);
	}

	public Boolean getBoolean(ByteBuffer bytes) {
		return (Boolean) convertToObject(this.boolConverter, bytes);
	}

	public ByteBuffer getBytes(Short value) {
		return convertToBytes(this.shortConverter, value);
	}

	public Short getShort(ByteBuffer bytes) {
		return (Short) convertToObject(this.shortConverter, bytes);
	}

	public ByteBuffer getBytes(Integer value) {
		return convertToBytes(this.intConverter, value);

	}

	public Integer getInteger(ByteBuffer bytes) {
		return (Integer) convertToObject(this.intConverter, bytes);
	}

	public ByteBuffer getBytes(Double value) {
		return convertToBytes(this.doubleConverter, value);
	}

	public Double getDouble(ByteBuffer bytes) {
		return (Double) convertToObject(this.doubleConverter, bytes);
	}

	public ByteBuffer getBytes(Long value) {
		return convertToBytes(this.longConverter, value);

	}

	public Long getLong(ByteBuffer bytes) {
		return (Long) convertToObject(this.longConverter, bytes);
	}

	public ByteBuffer getBytes(Float value) {
		return convertToBytes(this.floatConverter, value);
	}

	public Float getFloat(ByteBuffer bytes) {
		return (Float) convertToObject(this.floatConverter, bytes);
	}

	public ByteBuffer getBytes(String value) {
		return convertToBytes(this.stringConverter, value);
	}

	public String getString(ByteBuffer bytes) {
		return (String) convertToObject(this.stringConverter, bytes);
	}

	public ByteBuffer getBytes(Character value) {
		return convertToBytes(this.charConverter, value);
	}

	public Character getCharacter(ByteBuffer bytes) {
		return (Character) convertToObject(this.charConverter, bytes);
	}
	
	/**
     * Converts the backing array to the appropriate primitive.
     *
     * @return the deserialized primitive
     * @throws IllegalStateException if the underlying array does not contain the appropriate data
     */
    public static byte[] getArray(ByteBuffer buff) throws IllegalStateException {
        
    	if(buff == null)
    		return null;
    	
    	try {
        	buff.reset();
            return Arrays.copyOfRange(buff.array(), buff.position(), buff.limit());
        } catch (BufferUnderflowException e) {
            throw new IllegalStateException("Failed to read value due to invalid format.  See cause for details...", e);
        } finally {
        	buff.reset();
        }
    }
	
    public String convertToString(Object value) throws IOException {

    	ObjectStringConverter stringConverter = typeManager.getStringConverter(value.getClass());
    	if(stringConverter != null)
    		return stringConverter.toString(value);
    	
//        ByteArrayOutputStream bos = new ByteArrayOutputStream();
//        ObjectOutputStream oos = new ObjectOutputStream(bos);
//        oos.writeObject(value);
//        String name = new String(bos.toByteArray());
//        return name;
    	throw new NucleusDataStoreException(
				"Could not convert object to string.  Object types must be of any primitive, a UTF8 String, or java.util.UUID");

    }

	/**
	 * Allocate a byte buffer and convert the bytes with the given converter.
	 * Performs a mark and a reset on the internal buffer before invoking the
	 * write
	 * 
	 * @param converter
	 * @param value
	 * @return
	 */
	private ByteBuffer convertToBytes(IConverter converter, Object value) {
		ByteBuffer buff = converter.writeBytes(value, null, this);

		if (buff != null) {
			buff.limit();
			buff.reset();
		}		
//		System.out.println(String.format("Conversion Object -> Bytes >> value: %s ; hex: %s", value, new String(org.apache.commons.codec.binary.Hex.encodeHex(buff.array()))));

		return buff;
	}

	/**
	 * Write to the given buffer but do not rewind it.
	 * 
	 * @param converter
	 * @param value
	 * @return
	 */
	private ByteBuffer convertToBytes(IConverter converter, Object value,
			ByteBuffer buffer) {
		return converter.writeBytes(value, buffer, this);
	}

	/**
	 * Convert the bytes with the underlying buffer
	 * 
	 * @param converter
	 * @param bytes
	 * @return
	 */
	private Object convertToObject(IConverter converter, ByteBuffer bytes) {
		if (bytes == null) {
			return null;
		}
		return converter.getObject(bytes, this);
	}
	
	/**
	 * Check if the buffer has the remaining capacity to hold the number
	 * of bytes.  If not, create a new buffer with the required size and return it
	 * @param buffer
	 * @param size
	 * @return
	 */
	public static ByteBuffer check(ByteBuffer buffer, int size) {
		
		if(buffer == null){
			ByteBuffer newBuffer =  ByteBuffer.allocate(size);
			newBuffer.mark();
			return newBuffer;
		}
		
		if (buffer.remaining() < size) {

			int position = buffer.position();

			ByteBuffer newBuffer = ByteBuffer.allocate(position + size + 1);
			newBuffer.mark();

			buffer.reset();

			newBuffer.put(Arrays.copyOfRange(buffer.array(), 0, position));

			return newBuffer;

		}

		return buffer;
	}


}
